//
// Created by yfuxz on 2022/3/27.
//

#ifndef ICANLIVE_SAFEQUEUE_H
#define ICANLIVE_SAFEQUEUE_H



#include <pthread.h>
#include <queue>

using std::queue;

template<class T>
class SafeQueue {
    typedef void (*ReleaseHandle)(T &);

    typedef void (*SyncHandle)(queue<T> &);

public:
    SafeQueue() {
        pthread_mutex_init(&mutex, 0);
        pthread_cond_init(&cond, 0);
    }

    ~SafeQueue() {
        pthread_cond_destroy(&cond);
        pthread_mutex_destroy(&mutex);
    }

    void push(T val) {
        pthread_mutex_lock(&mutex);
        if (enable) {
            q.push(val);
            pthread_cond_signal(&cond);
        } else {
            releaseHandle(val);
        }
        pthread_mutex_unlock(&mutex);
    }

    int pop(T &value) {
        int ret = 0;
        pthread_mutex_lock(&mutex);
        while (enable && q.empty()) {
            pthread_cond_wait(&cond, &mutex);
        }
        if (!q.empty()) {
            value = q.front();
            q.pop();
            ret = 1;
        }
        pthread_mutex_unlock(&mutex);
        return ret;
    }

    void setEnable(bool enable) {
        pthread_mutex_lock(&mutex);
        this->enable = enable;
        pthread_cond_signal(&cond);
        pthread_mutex_unlock(&mutex);
    }

    bool empty() {
        return q.empty();
    }

    int size() {
        return q.size();
    }

    void clear() {
        pthread_mutex_lock(&mutex);
        int size = q.size();
        for (int i = 0; i < size; ++i) {
            T val = q.front();
            releaseHandle(val);
            q.pop();
        }
        pthread_mutex_unlock(&mutex);
    }

    void sync() {
        pthread_mutex_lock(&mutex);
        syncHandle(q);
        pthread_mutex_unlock(&mutex);
    }

    void setReleaseHandle(ReleaseHandle r) {
        releaseHandle = r;
    }

    void setSyncHandle(SyncHandle s) {
        syncHandle = s;
    }

private:
    pthread_mutex_t mutex;
    bool enable;
    queue<T> q;
    pthread_cond_t cond;
    ReleaseHandle releaseHandle;
    SyncHandle syncHandle;
};



#endif //ICANLIVE_SAFEQUEUE_H
